package com.central.dataManage.service.impl;

import com.alibaba.druid.sql.ast.SQLStatement;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.central.common.constant.CommonConstant;
import com.central.common.enums.DbType;
import com.central.common.enums.PrivilegeType;
import com.central.common.feign.ProjectService;
import com.central.common.feign.UserService;
import com.central.common.model.*;
import com.central.common.service.impl.SuperServiceImpl;
import com.central.common.utils.UserRoleUtil;
import com.central.dataManage.common.Constants;
import com.central.dataManage.common.datasourcePlugin.BaseDatasourcePlugin;
import com.central.dataManage.common.datasourcePlugin.DatasourcePluginFactory;
import com.central.dataManage.common.datasourcePlugin.KafkaDatasourcePlugin;
import com.central.dataManage.common.utils.PropertyUtils;
import com.central.dataManage.mapper.*;
import com.central.dataManage.model.*;
import com.central.dataManage.service.IDatabaseService;
import com.central.dataManage.service.IDatasourceService;
import com.central.dataManage.service.IDelCacheService;
import com.central.dataManage.service.ITableService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.acl.*;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;


/**
 * @author Tindy
 * @date 2021/6/24
 * @describe
 */
@Slf4j
@Service
public class TableServiceImpl extends SuperServiceImpl<TableInfoMapper, TableInfo> implements ITableService {

    @Autowired
    private UserService userService;
    @Autowired
    private ProjectService projectService;
    @Autowired
    private TableOwnerMapper tableOwnerMapper;
    @Autowired
    private TablePreviewMapper tablePreviewMapper;
    @Autowired
    private ProTablePrivilegeMapper proTablePrivilegeMapper;
    @Autowired
    private UserTablePrivilegeMapper userTablePrivilegeMapper;
    @Autowired
    private IDatasourceService datasourceService;
    @Autowired
    private IDatabaseService databaseService;
    @Autowired
    private IDelCacheService delCacheService;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void delByDatasourceId(Long datasourceId) {
        baseMapper.delete(new QueryWrapper<TableInfo>().eq("datasource_id", datasourceId));
    }

    @Override
    public SysUser findTableOwner(Long datasourceId, String dbName, String tableName) {
        TableOwner tableOwner = tableOwnerMapper.selectOne(new QueryWrapper<TableOwner>().eq("datasource_id", datasourceId).eq("db_name", dbName).eq("table_name", tableName));
        if (tableOwner != null) {
            return userService.selectById(tableOwner.getUserId());
        } else {
            return null;
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateTableOwner(Long datasourceId, String dbName, String tableName, Long userId) {
        //删除旧owner
        tableOwnerMapper.delete(new QueryWrapper<TableOwner>().eq("datasource_id", datasourceId).eq("db_name", dbName).eq("table_name", tableName));
        tableOwnerMapper.insert(new TableOwner(datasourceId, dbName, tableName, userId));
    }

    @Override
    public String findTablePreview(Long datasourceId, String dbName, String tableName) {
        TablePreview tablePreview = tablePreviewMapper.selectOne(new QueryWrapper<TablePreview>().eq("datasource_id", datasourceId).eq("db_name", dbName).eq("table_name", tableName));
        if (tablePreview == null) {
            return "{}";
        }
        return tablePreview.getPreviewData();
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result dropTable(Long tableId, String reason) {
        TableInfo tableInfo = getById(tableId);
        if (tableInfo != null) {
            DatasourceInfo datasourceInfo = datasourceService.getById(tableInfo.getDatasourceId());
            //删除表目录缓存
            DatabaseInfo databaseInfo = databaseService.getById(tableInfo.getDatabaseId());
            delCacheService.delTableList(databaseInfo.getId());
            if (datasourceInfo != null) {
                BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                try {
                    datasourcePlugin.dropTable(tableInfo.getDbName(), tableInfo.getTableName());
                    removeById(tableId);
                    return Result.succeed("删除成功");
                } catch (SQLException ex) {
                    return Result.failed("删除失败：" + ex);
                }
            } else {
                return Result.failed("删除失败：数据源不存在");
            }
        } else {
            return Result.failed("删除失败：数据表不存在");
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result createTable(CreateTableEntity createTableEntity) {
        DatabaseInfo databaseInfo = databaseService.getById(createTableEntity.getDatabaseId());
        if (databaseInfo != null) {
            DatasourceInfo datasourceInfo = datasourceService.getById(databaseInfo.getDatasourceId());
            if (datasourceInfo != null) {
                if(databaseInfo.getDatasourceType()==DbType.KAFKA) return createKafKaTable(datasourceInfo,databaseInfo,createTableEntity);
                BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                String ddl = null;
                String tableName = null;
                try {
                    switch (createTableEntity.getType()) {
                        case "sql":
                            ddl = createTableEntity.getDdl();
                            tableName = datasourcePlugin.getTableNameByDdl(ddl);
                            break;
                        case "visual":
                            TableInfo tableInfo = createTableEntity.getTableInfo();
                            ddl = datasourcePlugin.generateDdlSql(tableInfo);
                            tableName = tableInfo.getTableName();
                            break;
                        default:
                            return Result.failed("建表类型错误");
                    }
                    List<SQLStatement> sqlStatements = datasourcePlugin.checkCreateTableDdl(ddl);
                    datasourcePlugin.createTableByDdl(databaseInfo, sqlStatements);
                    tablePreviewMapper.insert(new TablePreview(datasourceInfo.getId(), databaseInfo.getDbName(), tableName, createTableEntity.getPreviewData()));
                    databaseService.synchDatabase(databaseInfo.getId());
                    return Result.succeed("创建成功");
                } catch (SQLException ex) {
                    return Result.failed("创建失败：" + ex);
                }
            } else {
                return Result.failed("创建失败：数据源不存在");
            }
        } else {
            return Result.failed("创建失败：数据库不存在");
        }
    }

    private Result createKafKaTable(DatasourceInfo datasourceInfo, DatabaseInfo databaseInfo, CreateTableEntity createTableEntity) {
        KafkaDatasourcePlugin datasourcePlugin = (KafkaDatasourcePlugin) DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
        AdminClient adminClient = datasourcePlugin.getAdminClient();
        TableInfo tableInfo = createTableEntity.getTableInfo();
        tableInfo.setDatasourceId(datasourceInfo.getId());
        tableInfo.setDatabaseId(databaseInfo.getId());
        tableInfo.setDbName(databaseInfo.getDbName());
        tableInfo.setDatasourceType(datasourceInfo.getDatasourceType());
        String topicName=databaseInfo.getDbName()+"."+tableInfo.getTableName();
        int numPartitions = StringUtils.isNotEmpty(tableInfo.getDdl())?Integer.valueOf(tableInfo.getDdl()):Constants.KAFKA_DEFAULT_PARTITIONS;
        tableInfo.setDdl(String.valueOf(numPartitions));
        short replicationFactor = Constants.KAFKA_DEFAULT_REPLICATION;
        if(numPartitions<=0 || replicationFactor<0) throw new IllegalArgumentException("numPartitions or replicationFactor 参数错误");
        ArrayList<NewTopic> newTopics = new ArrayList<>();
        NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) replicationFactor);
        newTopics.add(newTopic);
        try {
            adminClient.createTopics(newTopics).all().get();
            tablePreviewMapper.insert(new TablePreview(datasourceInfo.getId(), databaseInfo.getDbName(), tableInfo.getTableName(), createTableEntity.getPreviewData()));
            saveOrUpdate(tableInfo);
            return Result.succeed("创建成功");
        } catch (InterruptedException|ExecutionException e) {
            return Result.succeed("创建失败："+e);
        }finally {
            adminClient.close();
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Boolean synchTable(Long tableId) throws SQLException {
        boolean flag = false;
        TableInfo tableInfo = getById(tableId);
        if (tableInfo != null) {
            return databaseService.synchDatabase(tableInfo.getDatabaseId());
        }
        return flag;
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void delByDatabaseId(Long databaseId) {
        remove(new QueryWrapper<TableInfo>().eq("database_id", databaseId));
    }

    @Override
    public TableInfo findTableDetail(SysUser user, Long tableId) {
        TableInfo tableInfo = getById(tableId);
        //判断用户是否有表权限
        tableInfo = checkUserTabPerm(user, tableInfo);
        if (tableInfo != null) {
            String tablePreview = findTablePreview(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName());
            tableInfo.setPreviewData(tablePreview);
        }
        return tableInfo;
    }

    /**
     * 判断用户是否有表权限
     *
     * @param user
     * @param tableInfo
     */
    private TableInfo checkUserTabPerm(SysUser user, TableInfo tableInfo) {
        if (tableInfo != null) {
            //用户是否为全局数据管理员
            if (UserRoleUtil.checkGloDataAdmin(user.getRoles())) {
                return tableInfo;
            }
            //该表是否为该用户的用户库表
            DatabaseInfo userDb = databaseService.getUserDb(user);
            Boolean isUserTable = userDb != null && tableInfo.getDatabaseId() == userDb.getId();
            if (isUserTable) return tableInfo;
            //该表是否为该用户当前项目的项目库表
            DatabaseInfo proDb = databaseService.getProDb(user);
            Boolean isProTable = proDb != null && tableInfo.getDatabaseId() == proDb.getId();
            if (isProTable) return tableInfo;
            //该用户对该表的权限不为空
            UserTablePrivilege userTablePrivilege = userTablePrivilegeMapper.selectOne(new QueryWrapper<UserTablePrivilege>().eq("datasource_id", tableInfo.getDatasourceId())
                    .eq("db_name", tableInfo.getDbName()).eq("table_name", tableInfo.getTableName())
                    .eq("pro_id", user.getCurrentProjectId()).eq("user_id", user.getId()));
            if (userTablePrivilege != null && !userTablePrivilege.getPrivilege().isEmpty()) return tableInfo;
        }
        return null;
    }

    @Override
    public Result<List<ProjectPrivileges>> findProjectPrivileges(Long tableId) {
        TableInfo tableInfo = getById(tableId);
        if (tableInfo == null) return Result.failed("数据表不存在");
        List<ProTablePrivilege> proTablePrivileges = proTablePrivilegeMapper.selectList(new QueryWrapper<ProTablePrivilege>().eq("datasource_id", tableInfo.getDatasourceId()).eq("db_name", tableInfo.getDbName()).eq("table_name", tableInfo.getTableName()));
        List<Long> proIds = proTablePrivileges.stream().map(t -> t.getProId()).collect(Collectors.toList());
        List<SysProject> projects = projectService.findProByProIds(proIds);
        HashMap<Long, String> projectMap = new HashMap();
        for (SysProject project : projects) {
            projectMap.put(project.getId(), project.getName());
        }
        List<ProjectPrivileges> projectPrivilegesList = new ArrayList<>();
        for (ProTablePrivilege proTablePrivilege : proTablePrivileges) {
            ProjectPrivileges projectPrivileges = new ProjectPrivileges(proTablePrivilege.getProId(), projectMap.get(proTablePrivilege.getProId()), proTablePrivilege.getPrivilege());
            projectPrivilegesList.add(projectPrivileges);
        }
        return Result.succeed(projectPrivilegesList);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result addProjectPrivileges(Long tableId, Long proId, List<PrivilegeType> privileges) throws SQLException {
        TableInfo tableInfo = getById(tableId);
        if (tableInfo == null) return Result.failed("数据表不存在");
        DatasourceInfo datasourceInfo = datasourceService.getById(tableInfo.getDatasourceId());
        //如果是系统hive库，需判断权限是否在系统hive所支持的dml操作范围内
        if (datasourceInfo.getIsSys() && datasourceInfo.getDatasourceType() == DbType.HIVE) {
            String dmlOperationStr = PropertyUtils.getString(Constants.SYS_HIVE_DML_OPERATIONS, Constants.SYS_HIVE_DML_OPERATIONS_DEFAULT);
            List dmlOperations = Arrays.asList(dmlOperationStr.toUpperCase().split(","));
            for (PrivilegeType privilege : privileges) {
                if (!dmlOperations.contains(privilege.name())) {
                    return Result.failed("系统hive库不支持：" + privilege.name());
                }
            }
        }
        //如果修改权限，将权限范围缩小，需要删除该项目所有用户的对于表权限
        ProTablePrivilege oldProTablePrivilege = proTablePrivilegeMapper.selectOne(new QueryWrapper<ProTablePrivilege>()
                .eq("datasource_id", tableInfo.getDatasourceId())
                .eq("db_name", tableInfo.getDbName())
                .eq("table_name", tableInfo.getTableName())
                .eq("pro_id", proId));
        if (oldProTablePrivilege != null) {
            List<PrivilegeType> oldPrivileges = oldProTablePrivilege.getPrivilege();
            //被缩小的权限
            oldPrivileges.removeAll(privileges);
            if (oldPrivileges.size() > 0) {
                List<UserTablePrivilege> userTablePrivileges = userTablePrivilegeMapper.selectList(new QueryWrapper<UserTablePrivilege>()
                        .eq("datasource_id", tableInfo.getDatasourceId())
                        .eq("db_name", tableInfo.getDbName())
                        .eq("table_name", tableInfo.getTableName())
                        .eq("pro_id", proId));
                for (UserTablePrivilege userTablePrivilege : userTablePrivileges) {
                    userTablePrivilege.getPrivilege().removeAll(oldPrivileges);
                    userTablePrivilegeMapper.update(userTablePrivilege, new QueryWrapper<UserTablePrivilege>()
                            .eq("datasource_id", tableInfo.getDatasourceId())
                            .eq("db_name", tableInfo.getDbName())
                            .eq("table_name", tableInfo.getTableName())
                            .eq("pro_id", proId)
                            .eq("user_id", userTablePrivilege.getUserId()));
                }
                //如果是系统hive/kafka库，需要修改相应的sentry权限表
                if (datasourceInfo.getIsSys()) {
                    switch (datasourceInfo.getDatasourceType()) {
                        case HIVE:
                            BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                            Connection conn = datasourcePlugin.getConn();
                            String delPrivileges = StringUtils.join(oldPrivileges, ',');
                            for (UserTablePrivilege userTablePrivilege : userTablePrivileges) {
                                //删除项目缩小的权限
                                conn.createStatement().execute("REVOKE " + delPrivileges + " ON TABLE " + userTablePrivilege.getDbName() + "." + userTablePrivilege.getTableName()
                                        + " FROM ROLE " + CommonConstant.KA_PREFIX + userTablePrivilege.getProName() + "_" + userTablePrivilege.getUsername() + "_role");
                            }
                            conn.close();
                            break;
                        case KAFKA:
                            ArrayList<AclBindingFilter> list = new ArrayList<>();
                            for (PrivilegeType oldPrivilege : oldPrivileges) {
                                for (UserTablePrivilege userTablePrivilege : userTablePrivileges) {
                                    String group = CommonConstant.KA_PREFIX + userTablePrivilege.getProName() + "_" + userTablePrivilege.getUsername();
                                    String role = group + "_role";
                                    ResourcePatternFilter resourcePatternFilter = new ResourcePatternFilter(ResourceType.TOPIC, userTablePrivilege.getDbName() + "." + userTablePrivilege.getTableName(), PatternType.LITERAL);
                                    AccessControlEntryFilter accessControlEntryFilter = null;
                                    switch (oldPrivilege) {
                                        case SELECT:
                                            accessControlEntryFilter = new AccessControlEntryFilter(group + ":" + role, "*", AclOperation.READ, AclPermissionType.ALLOW);
                                            list.add(new AclBindingFilter(resourcePatternFilter, accessControlEntryFilter));
                                            break;
                                        case INSERT:
                                            accessControlEntryFilter = new AccessControlEntryFilter(group + ":" + role, "*", AclOperation.WRITE, AclPermissionType.ALLOW);
                                            list.add(new AclBindingFilter(resourcePatternFilter, accessControlEntryFilter));
                                            break;
                                        default:
                                            break;
                                    }
                                }
                            }
                            if (list.isEmpty()) break;
                            KafkaDatasourcePlugin kafkaDatasourcePlugin = (KafkaDatasourcePlugin) DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                            AdminClient adminClient = kafkaDatasourcePlugin.getAdminClient();
                            try {
                                adminClient.deleteAcls(list).all().get();
                                adminClient.close();
                            } catch (InterruptedException | ExecutionException e) {
                                log.error(e.getLocalizedMessage());
                            }
                            break;
                        default:
                            break;
                    }

                }
            }
        }
        //删除旧权限
        proTablePrivilegeMapper.delete(new QueryWrapper<ProTablePrivilege>()
                .eq("datasource_id", tableInfo.getDatasourceId())
                .eq("db_name", tableInfo.getDbName())
                .eq("table_name", tableInfo.getTableName())
                .eq("pro_id", proId));
        if (!privileges.isEmpty()) {
            proTablePrivilegeMapper.insert(new ProTablePrivilege(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName(), proId, privileges));
        }
        return Result.succeed("操作成功");
    }

    @Override
    public Result<List<UserPrivileges>> findUserPrivileges(Long proId, Long tableId) {
        TableInfo tableInfo = getById(tableId);
        if (tableInfo == null) return Result.failed("数据表不存在");
        List<UserTablePrivilege> userTablePrivileges = userTablePrivilegeMapper.selectList(new QueryWrapper<UserTablePrivilege>()
                .eq("datasource_id", tableInfo.getDatasourceId())
                .eq("db_name", tableInfo.getDbName())
                .eq("table_name", tableInfo.getTableName())
                .eq("pro_id", proId));
        HashMap<Long, List<PrivilegeType>> userMap = new HashMap();
        for (UserTablePrivilege userTablePrivilege : userTablePrivileges) {
            userMap.put(userTablePrivilege.getUserId(), userTablePrivilege.getPrivilege());
        }

        List<SysUser> users = userService.findUsersOnPro(1, 1000, proId).getData();
        List<UserPrivileges> userPrivilegesList = new ArrayList<>();
        for (SysUser user : users) {
            userPrivilegesList.add(new UserPrivileges(user.getId(), user.getUsername(), userMap.get(user.getId())));
        }
        return Result.succeed(userPrivilegesList);
    }

    @Transactional
    @Override
    public Result addUserPrivileges(SysUser gdbUser, Long userId, Long tableId, List<PrivilegeType> privileges) throws SQLException, ExecutionException, InterruptedException {
        TableInfo tableInfo = getById(tableId);
        if (tableInfo == null) return Result.failed("数据表不存在");
        //获取项目表权限
        //如果该表为当前项目库表，则项目表权限为系统hive库dml所有权限
        DatabaseInfo databaseInfo = databaseService.getById(tableInfo.getDatabaseId());
        DatasourceInfo datasourceInfo = datasourceService.getById(tableInfo.getDatasourceId());
        ProTablePrivilege proTablePrivilege = null;
        String dmlOperationStr = PropertyUtils.getString(Constants.SYS_HIVE_DML_OPERATIONS, Constants.SYS_HIVE_DML_OPERATIONS_DEFAULT).toUpperCase();
        if (datasourceInfo.getIsSys() && databaseInfo.getDatasourceType() == DbType.HIVE && databaseInfo.getDbName().equals(CommonConstant.PRO_DB_PREFIX + gdbUser.getProName())) {
            List dmlOperations = Arrays.stream(dmlOperationStr.toUpperCase().split(",")).map(x -> PrivilegeType.valueOf(x)).collect(Collectors.toList());
            proTablePrivilege = new ProTablePrivilege(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName(), gdbUser.getCurrentProjectId(), dmlOperations);
        } else {
            proTablePrivilege = proTablePrivilegeMapper.selectOne(new QueryWrapper<ProTablePrivilege>()
                    .eq("datasource_id", tableInfo.getDatasourceId())
                    .eq("db_name", tableInfo.getDbName())
                    .eq("table_name", tableInfo.getTableName())
                    .eq("pro_id", gdbUser.getCurrentProjectId()));
        }

        if (proTablePrivilege == null || (proTablePrivilege != null && !proTablePrivilege.getPrivilege().containsAll(privileges))) {
            return Result.failed("授权失败：用户授权超过项目表权限范围");
        }
        //删除旧权限
        userTablePrivilegeMapper.delete(new QueryWrapper<UserTablePrivilege>()
                .eq("datasource_id", tableInfo.getDatasourceId())
                .eq("db_name", tableInfo.getDbName())
                .eq("table_name", tableInfo.getTableName())
                .eq("pro_id", gdbUser.getCurrentProjectId())
                .eq("user_id", userId));
        SysProject project = projectService.findById(gdbUser.getCurrentProjectId());
        SysUser user = userService.selectById(userId);
        if (!privileges.isEmpty()) {
            userTablePrivilegeMapper.insert(new UserTablePrivilege(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName(), gdbUser.getCurrentProjectId(), project.getName(), userId, user.getUsername(), privileges));
        }
        //如果数据源是系统hive库，更新对应sentry权限表
        if (datasourceInfo.getIsSys()) {
            switch (databaseInfo.getDatasourceType()) {
                case HIVE:
                    BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                    Connection conn = datasourcePlugin.getConn();
                    //删除旧权限
                    conn.createStatement().execute("REVOKE " + dmlOperationStr + " ON TABLE " + tableInfo.getDbName() + "." + tableInfo.getTableName()
                            + " FROM ROLE " + CommonConstant.KA_PREFIX + project.getName() + "_" + user.getUsername() + "_role");
                    //添加新权限
                    if (!privileges.isEmpty()) {
                        conn.createStatement().execute("GRANT " + StringUtils.join(privileges, ',') + " ON TABLE " + tableInfo.getDbName() + "." + tableInfo.getTableName()
                                + " TO ROLE " + CommonConstant.KA_PREFIX + project.getName() + "_" + user.getUsername() + "_role");
                    }
                    conn.close();
                    break;
                case KAFKA:
                    String group = CommonConstant.KA_PREFIX + project.getName() + "_" + user.getUsername();
                    String role = CommonConstant.KA_PREFIX + project.getName() + "_" + user.getUsername() + "_role";
                    KafkaDatasourcePlugin kafkaDatasourcePlugin = (KafkaDatasourcePlugin) DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                    AdminClient adminClient = kafkaDatasourcePlugin.getAdminClient();
                    //删除旧权限
                    List<AclBindingFilter> list1 = new ArrayList<>();
                    ResourcePatternFilter resourcePatternFilter = new ResourcePatternFilter(ResourceType.TOPIC, tableInfo.getDbName() + "." + tableInfo.getTableName(), PatternType.LITERAL);
                    AccessControlEntryFilter accessControlEntryFilter = new AccessControlEntryFilter(group + ":" + role, "*", AclOperation.ALL, AclPermissionType.ALLOW);
                    AclBindingFilter aclBindingFilter = new AclBindingFilter(resourcePatternFilter, accessControlEntryFilter);
                    list1.add(aclBindingFilter);
                    adminClient.deleteAcls(list1).all().get();
                    //添加新权限
                    List<AclBinding> list2 = new ArrayList<>();
                    ResourcePattern resourcePattern = null;
                    AccessControlEntry accessControlEntry = null;
                    for (PrivilegeType privilege : privileges) {
                        switch (privilege) {
                            case SELECT:
                                resourcePattern = new ResourcePattern(ResourceType.TOPIC, tableInfo.getDbName() + "." + tableInfo.getTableName(), PatternType.LITERAL);
                                accessControlEntry = new AccessControlEntry(group + ":" + role, "*", AclOperation.READ, AclPermissionType.ALLOW);
                                list2.add(new AclBinding(resourcePattern, accessControlEntry));
                                resourcePattern = new ResourcePattern(ResourceType.TOPIC, tableInfo.getDbName() + "." + tableInfo.getTableName(), PatternType.LITERAL);
                                accessControlEntry = new AccessControlEntry(group + ":" + role, "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW);
                                list2.add(new AclBinding(resourcePattern, accessControlEntry));
                                resourcePattern = new ResourcePattern(ResourceType.GROUP, "*", PatternType.LITERAL);
                                accessControlEntry = new AccessControlEntry(group + ":" + role, "*", AclOperation.ALL, AclPermissionType.ALLOW);
                                list2.add(new AclBinding(resourcePattern, accessControlEntry));
                                break;
                            case INSERT:
                                resourcePattern = new ResourcePattern(ResourceType.TOPIC, tableInfo.getDbName() + "." + tableInfo.getTableName(), PatternType.LITERAL);
                                accessControlEntry = new AccessControlEntry(group + ":" + role, "*", AclOperation.WRITE, AclPermissionType.ALLOW);
                                list2.add(new AclBinding(resourcePattern, accessControlEntry));
                                resourcePattern = new ResourcePattern(ResourceType.TOPIC, tableInfo.getDbName() + "." + tableInfo.getTableName(), PatternType.LITERAL);
                                accessControlEntry = new AccessControlEntry(group + ":" + role, "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW);
                                list2.add(new AclBinding(resourcePattern, accessControlEntry));
                                break;
                            default:
                                break;
                        }
                    }
                    adminClient.createAcls(list2).all().get();
                    break;
                default:
                    break;
            }

        }
        return Result.succeed("操作成功");
    }

    @Override
    public Result<String> format(Long datasourceId, String sql) {
        DatasourceInfo datasourceInfo = datasourceService.getById(datasourceId);
        BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
        return Result.succeed(datasourcePlugin.format(sql));
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result modifyTable(ModifyTableEntity modifyTableEntity) throws ExecutionException, InterruptedException {
        TableInfo tableInfo = getById(modifyTableEntity.getTableId());
        if (tableInfo != null) {
            DatabaseInfo databaseInfo = databaseService.getById(tableInfo.getDatabaseId());
            DatasourceInfo datasourceInfo = datasourceService.getById(tableInfo.getDatasourceId());
            if (databaseInfo != null && datasourceInfo != null) {
                //如果是kafka数据源，直接修改元数据
                if(datasourceInfo.getDatasourceType()==DbType.KAFKA) return modifyKafkaTable(datasourceInfo,databaseInfo,tableInfo,modifyTableEntity);
                BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
                String ddl = null;
                try {
                    switch (modifyTableEntity.getType()) {
                        case "sql":
                            ddl = modifyTableEntity.getDdl();
                            break;
                        case "visual":
                            TableInfo tableInfo2 = modifyTableEntity.getTableInfo();
                            ddl = datasourcePlugin.generateModifySql(tableInfo, tableInfo2);
                            break;
                        default:
                            return Result.failed("修改类型错误");
                    }
                    List<SQLStatement> sqlStatements = datasourcePlugin.checkModifyTableDdl(ddl, tableInfo.getTableName());
                    datasourcePlugin.modifyTableByDdl(databaseInfo, sqlStatements);
                    tablePreviewMapper.delete(new QueryWrapper<TablePreview>()
                            .eq("datasource_id", tableInfo.getDatasourceId())
                            .eq("db_name", tableInfo.getDbName())
                            .eq("table_name", tableInfo.getTableName()));
                    tablePreviewMapper.insert(new TablePreview(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName(), modifyTableEntity.getPreviewData()));
                    databaseService.synchDatabase(databaseInfo.getId());
                    return Result.succeed("修改成功");
                } catch (Exception ex) {
                    return Result.failed("修改失败：" + ex);
                }
            } else {
                return Result.failed("修改失败：数据源/库不存在");
            }
        } else {
            return Result.failed("修改失败：数据表不存在");
        }
    }

    private Result modifyKafkaTable(DatasourceInfo datasourceInfo, DatabaseInfo databaseInfo, TableInfo tableInfo, ModifyTableEntity modifyTableEntity) throws ExecutionException, InterruptedException {
        int oldNumPartitions=Integer.valueOf(tableInfo.getDdl());
        int newNumPartitions=StringUtils.isNotEmpty(modifyTableEntity.getDdl())?Integer.valueOf(modifyTableEntity.getDdl()):1;
        if(newNumPartitions>oldNumPartitions){
            KafkaDatasourcePlugin datasourcePlugin = (KafkaDatasourcePlugin) DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
            AdminClient adminClient = datasourcePlugin.getAdminClient();
            Map<String, NewPartitions> map=new HashMap<>();
            map.put(databaseInfo.getDbName()+"."+tableInfo.getTableName(),NewPartitions.increaseTo(newNumPartitions));
            adminClient.createPartitions(map).all().get();
            adminClient.close();
        }else {
            modifyTableEntity.getTableInfo().setDdl(tableInfo.getDdl());
        }
        modifyTableEntity.getTableInfo().setId(tableInfo.getId());
        modifyTableEntity.getTableInfo().setDatasourceId(tableInfo.getDatasourceId());
        modifyTableEntity.getTableInfo().setDatabaseId(tableInfo.getDatabaseId());
        modifyTableEntity.getTableInfo().setDatasourceType(tableInfo.getDatasourceType());
        modifyTableEntity.getTableInfo().setDbName(tableInfo.getDbName());
        modifyTableEntity.getTableInfo().setTableName(tableInfo.getTableName());
        updateById(modifyTableEntity.getTableInfo());
        tablePreviewMapper.delete(new QueryWrapper<TablePreview>()
                .eq("datasource_id", tableInfo.getDatasourceId())
                .eq("db_name", tableInfo.getDbName())
                .eq("table_name", tableInfo.getTableName()));
        tablePreviewMapper.insert(new TablePreview(tableInfo.getDatasourceId(), tableInfo.getDbName(), tableInfo.getTableName(), modifyTableEntity.getPreviewData()));
        return Result.succeed("修改成功");
    }

    @Override
    public List<String> findDataTypes(Long databaseId) {
        DatabaseInfo databaseInfo = databaseService.getById(databaseId);
        DatasourceInfo datasourceInfo = datasourceService.getById(databaseInfo.getDatasourceId());
        BaseDatasourcePlugin datasourcePlugin = DatasourcePluginFactory.getDatasourcePlugin(datasourceInfo);
        return datasourcePlugin.getDataTypes();
    }

    @Override
    public Map<String, List<PrivilegeType>> getAllTablePrivilege(Long datasourceId, Long proId, Long userId, Set<String> tables) {
        List<UserTablePrivilege> all = userTablePrivilegeMapper.getAllTablePrivilege(datasourceId,proId,userId,tables);
        HashMap<String, List<PrivilegeType>> map = new HashMap<>();
        for (UserTablePrivilege userTablePrivilege : all) {
            String key = datasourceId + "." + userTablePrivilege.getDbName() + "." + userTablePrivilege.getTableName();
            map.put(key.toLowerCase(), userTablePrivilege.getPrivilege());
        }
        return map;
    }

    @Override
    public TableInfo getTableInfo(Long datasourceId, String dbName, String tableName) {
        return baseMapper.selectOne(new QueryWrapper<TableInfo>()
                .eq("datasource_id",datasourceId)
                .eq("db_name",dbName)
                .eq("table_name",tableName));
    }
}